package org.tao.lightningmq.broker.event.spi.listenr;

import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.event.model.PushMsgEvent;
import org.tao.lightningmq.broker.model.TxMessageAckModel;
import org.tao.lightningmq.broker.timewheel.DelayMessageDTO;
import org.tao.lightningmq.broker.timewheel.SlotStoreTypeEnum;
import org.tao.lightningmq.common.dto.MessageDTO;
import org.tao.lightningmq.common.dto.SendMessageToBrokerResponseDTO;
import org.tao.lightningmq.common.dto.TxMessageDTO;
import org.tao.lightningmq.common.enums.*;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.io.IOException;

/**
 * @Author lht
 * @date 2025/1/20 - 14:09
 * @description: 生产者推送消息事件监听器
 */
public class PushMsgListener implements Listener<PushMsgEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(PushMsgListener.class);

    @Override
    public void onReceive(PushMsgEvent event) throws Exception {
        MessageDTO messageDTO = event.getMessageDTO();
        boolean isDelayMsg = messageDTO.getDelay() > 0;
        boolean isHalfMsg = messageDTO.getTxFlag() == TxMessageFlagEnum.HALF_MSG.getCode();
        boolean isRemainHalfAck = messageDTO.getTxFlag() == TxMessageFlagEnum.REMAIN_HALF_ACK.getCode();
        if (isDelayMsg) {
            appendDelayMsgHandler(messageDTO, event);
        } else if (isHalfMsg) {
            halfMsgHandler(messageDTO, event);
        } else if (isRemainHalfAck) {
            remainHalfMsgAckHandler(messageDTO, event);
        } else {
            appendDefaultMsgHandler(messageDTO, event);
        }
    }

    private void remainHalfMsgAckHandler(MessageDTO messageDTO, PushMsgEvent event) throws IOException {
        LocalTransactionState localTransactionState = LocalTransactionState.of(messageDTO.getLocalTxState());
        if (localTransactionState == LocalTransactionState.COMMIT) {
            CommonCache.getTxMessageAckModelMap().remove(messageDTO.getMsgId());
            CommonCache.getCommitLogAppendHandler().appendMsg(messageDTO);
            LOGGER.info("收到事务消息的commit请求");
        } else if (localTransactionState == LocalTransactionState.ROLLBACK) {
            CommonCache.getTxMessageAckModelMap().remove(messageDTO.getMsgId());
            LOGGER.info("收到事务消息的rollback请求");
        }
        SendMessageToBrokerResponseDTO sendMsgResp = new SendMessageToBrokerResponseDTO();
        sendMsgResp.setMsgId(messageDTO.getMsgId());
        sendMsgResp.setStatus(SendMessageToBrokerResponseStatus.SUCCESS.getCode());
        sendMsgResp.setDesc("send tx remain ack msg success");
        TcpMsg responseMsg = new TcpMsg(BrokerResponseCode.REMAIN_ACK_MSG_SEND_SUCCESS.getCode(), com.alibaba.fastjson.JSON.toJSONBytes(sendMsgResp));
        event.getContext().writeAndFlush(responseMsg);
    }

    private void halfMsgHandler(MessageDTO messageDTO, PushMsgEvent event) {
        TxMessageAckModel txMessageAckModel = new TxMessageAckModel();
        txMessageAckModel.setMessageDTO(messageDTO);
        txMessageAckModel.setChannelHandlerContext(event.getContext());
        txMessageAckModel.setFirstSendTime(System.currentTimeMillis());
        CommonCache.getTxMessageAckModelMap().put(messageDTO.getMsgId(), txMessageAckModel);
        // 推送到时间轮
        TxMessageDTO txMessageDTO = new TxMessageDTO();
        txMessageDTO.setMsgId(messageDTO.getMsgId());
        DelayMessageDTO delayMessageDTO = new DelayMessageDTO();
        delayMessageDTO.setData(txMessageDTO);
        delayMessageDTO.setSlotStoreType(SlotStoreTypeEnum.TX_MESSAGE_DTO);
        delayMessageDTO.setNextExecuteTime(System.currentTimeMillis() + 3 * 1000);
        delayMessageDTO.setDelay(3);
        CommonCache.getTimeWheelModelManager().addDelayMessage(delayMessageDTO);
        // 回复客户端
        SendMessageToBrokerResponseDTO sendMsgResp = new SendMessageToBrokerResponseDTO();
        sendMsgResp.setMsgId(messageDTO.getMsgId());
        sendMsgResp.setStatus(SendMessageToBrokerResponseStatus.SUCCESS.getCode());
        sendMsgResp.setDesc("send tx half msg success");
        TcpMsg responseMsg = new TcpMsg(BrokerResponseCode.HALF_MSG_SEND_SUCCESS.getCode(), com.alibaba.fastjson.JSON.toJSONBytes(sendMsgResp));
        event.getContext().writeAndFlush(responseMsg);
    }

    private void appendDelayMsgHandler(MessageDTO messageDTO, PushMsgEvent event) throws Exception {
        int delay = messageDTO.getDelay();
        if (delay >= 3600) {
            throw new IllegalArgumentException("delay time must less than 3600s");
        }
        DelayMessageDTO delayMessageDTO = new DelayMessageDTO();
        delayMessageDTO.setDelay(messageDTO.getDelay());
        delayMessageDTO.setData(messageDTO);
        delayMessageDTO.setSlotStoreType(SlotStoreTypeEnum.DELAY_MESSAGE_DTO);
        delayMessageDTO.setNextExecuteTime(System.currentTimeMillis() + delay * 1000L);
        CommonCache.getTimeWheelModelManager().addDelayMessage(delayMessageDTO);
        // 将延迟消息做一个持久化 TODO 持久化后在启动时需要检查这个队列
//        MessageDTO delayMessage = new MessageDTO();
//        delayMessage.setBody(com.alibaba.fastjson.JSON.toJSONBytes(delayMessageDTO));
//        delayMessage.setTopic("delay_queue");
//        delayMessage.setQueueId(0);
//        delayMessage.setSendWay(MessageSendWay.ASYNC.getCode());
//        CommonCache.getCommitLogAppendHandler().appendMsg(delayMessage, event);
        SendMessageToBrokerResponseDTO sendMsgResp = new SendMessageToBrokerResponseDTO();
        sendMsgResp.setMsgId(messageDTO.getMsgId());
        sendMsgResp.setStatus(SendMessageToBrokerResponseStatus.SUCCESS.getCode());
        sendMsgResp.setDesc("send delay msg success");
        TcpMsg tcpMsg = new TcpMsg(BrokerResponseCode.SEND_MSG_RESP.getCode(), JSON.toJSONBytes(sendMsgResp));
        event.getContext().writeAndFlush(tcpMsg);
    }

    private void appendDefaultMsgHandler(MessageDTO messageDTO, PushMsgEvent event) {
        try {
            CommonCache.getCommitLogAppendHandler().appendMsg(messageDTO, event);
        } catch (Exception e) {
            LOGGER.warn("appendDefaultMsgHandler error {}", e);
            SendMessageToBrokerResponseDTO resp = new SendMessageToBrokerResponseDTO();
            resp.setStatus(SendMessageToBrokerResponseStatus.FAIL.getCode());
            resp.setMsgId(messageDTO.getMsgId());
            TcpMsg tcpMsg = new TcpMsg(BrokerResponseCode.SEND_MSG_RESP.getCode(), JSON.toJSONBytes(resp));
            event.getContext().writeAndFlush(tcpMsg);
        }
    }
}
